// Copyright 2019 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ------------------------------------------------------------------
// Modification history:
// feature: listener code decoupling
// feature: provide query interfaces
// feature: develop dispatcher
// feature: Split history as an independent common module
// feature: Unified Transport Abstraction Layer Unified Transport Abstraction Layer
// feature: discovery support client and server
// ------------------------------------------------------------------

#include <discovery/participant/PDPListener.h>

#include <discovery/endpoint/EDP.h>
#include <discovery/participant/PDP.h>
#include <history/ReaderHistory.h>
#include <edds/rtps/participant/ParticipantDiscoveryInfo.h>
#include <history/ParticipantListener.h>
#include <edds/rtps/reader/RTPSReader.h>
#include <edds/rtps/resources/TimedEvent.h>

#include <ertps/utils/TimeConversion.h>

#include <mutex>

#include <elog/Log.hpp>

#include <edds/dds/core/policy/ParameterList.hpp>
#include <rtps/network/ExternalLocatorsProcessor.hpp>
#include <rtps/participant/RTPSParticipantImpl.h>

using ParameterList = evbs::edds::dds::ParameterList;

namespace evbs {
namespace ertps {
namespace rtps {

PDPListener::PDPListener(PDP* parent)
    : parent_pdp_(parent),
      temp_participant_data_(parent->getRTPSParticipant()->getRTPSParticipantAttributes().allocation) {}

void PDPListener::onNewCacheChangeAdded(vbs::dispatcher::SubstrateReader* sub_reader,
                                        const CacheChange_t* const change_in) {
    CacheChange_t* change = const_cast<CacheChange_t*>(change_in);
    RTPSReader* reader = static_cast<RTPSReader*>(sub_reader);
    GUID_t writer_guid = change->writerGUID;
    logDebug(RTPS_PDP, "SPDP Message received from: " << writer_guid);
    std::shared_ptr<ParticipantProxyData> ppd_ptr;

    // Make sure we have an instance handle (i.e GUID)
    if (change->instanceHandle == c_InstanceHandle_Unknown) {
        if (!this->get_key(change)) {
            logWarning(RTPS_PDP, writer_guid << " Problem getting the key of the change, removing");
            (void)parent_pdp_->builtin_endpoints_->remove_from_pdp_reader_history(change);
            return;
        }
    }

    // Take GUID from instance handle
    GUID_t guid;
    vbsutil::xmlparser::iHandle2GUID(guid, change->instanceHandle);

    if (change->kind == ALIVE) {
        // Ignore announcement from own RTPSParticipant
        if (guid == parent_pdp_->getRTPSParticipant()->getGuid()) {
            (void)parent_pdp_->builtin_endpoints_->remove_from_pdp_reader_history(change);
            logDebug(RTPS_PDP, "Ignore announcement from own RTPSParticipant, removing. writer:" << writer_guid);
            return;
        }

        // Release reader lock to avoid ABBA lock. PDP mutex should always be first.
        // Keep change information on local variables to check consistency later
        SequenceNumber_t seq_num = change->sequenceNumber;
        reader->getMutex().unlock();
        std::unique_lock<std::recursive_mutex> lock(*parent_pdp_->getMutex());
        reader->getMutex().lock();

        // If change is not consistent, it will be processed on the thread that has overriten it
        if ((ALIVE != change->kind) || (seq_num != change->sequenceNumber) || (writer_guid != change->writerGUID)) {
            logDebug(RTPS_PDP, writer_guid << " Ignore change is not consistent.");
            return;
        }

        // Access to temp_participant_data_ is protected by reader lock

        // Load information on temp_participant_data_
        CDRMessage_t msg(change->serializedPayload);
        temp_participant_data_.clear();
        if (temp_participant_data_.readFromCDRMessage(&msg, true, parent_pdp_->getRTPSParticipant())) {
            // After correctly reading it
            change->instanceHandle = temp_participant_data_.m_key;
            guid = temp_participant_data_.m_guid;

            // For dynamic GUIDs without PID_HOSTID_PID_INFO, extract host_id and pid from the GUID.
            if (!guid.guidPrefix.is_guid_static() && temp_participant_data_.m_hostId == 0 &&
                temp_participant_data_.m_pid == 0) {
                temp_participant_data_.m_hostId = guid.guidPrefix.get_host_id();
                temp_participant_data_.m_pid = guid.guidPrefix.get_process_id();
            }

            if (parent_pdp_->getRTPSParticipant()->is_participant_ignored(guid.guidPrefix)) {
                logDebug(RTPS_PDP, writer_guid << " Ignore participant: " << guid.guidPrefix << ".");
                return;
            }

            if (!check_discovery_conditions(temp_participant_data_)) {
                logDebug(RTPS_PDP, writer_guid << " check discovery conditions failed.");
                return;
            }

            // Filter locators
            const auto& pattr = parent_pdp_->getRTPSParticipant()->getAttributes();
            edds::rtps::ExternalLocatorsProcessor::filter_remote_locators(
                temp_participant_data_, pattr.builtin.metatraffic_external_unicast_locators,
                pattr.default_external_unicast_locators, pattr.ignore_non_matching_locators);

            for (ParticipantProxyData* it : parent_pdp_->participant_deleting_proxies_) {
                if (guid == it->m_guid) {
                    elogWarning(RTPS_PDP, guid << " is in process of timeout deleting.");
                    return;
                }
            }

            // Check if participant already exists (updated info)
            bool already_processed = false;
            ParticipantProxyData* pdata = nullptr;
            for (ParticipantProxyData* it : parent_pdp_->participant_proxies_) {
                if (guid == it->m_guid) {
                    pdata = it;

                    if (it->m_sample_identity.writer_guid() == change->writerGUID &&
                        it->m_sample_identity.sequence_number() == change->sequenceNumber) {
                        already_processed = true;
                    }

                    break;
                }
            }
            // Only process the DATA(p) if it is not a repeated one
            if (!already_processed) {
                temp_participant_data_.m_sample_identity.writer_guid(change->writerGUID);
                temp_participant_data_.m_sample_identity.sequence_number(change->sequenceNumber);
                process_alive_data(pdata, temp_participant_data_, writer_guid, reader, lock);
            }
        }
    } else if (reader->matched_writer_is_matched(writer_guid)) {
        reader->getMutex().unlock();
        if (parent_pdp_->remove_remote_participant(guid, ParticipantDiscoveryInfo::REMOVED_PARTICIPANT)) {
            reader->getMutex().lock();
            // All changes related with this participant have been removed from history by
            // remove_remote_participant
            return;
        }
        reader->getMutex().lock();
    }

    // Remove change form history.
    parent_pdp_->builtin_endpoints_->remove_from_pdp_reader_history(change);
}

bool PDPListener::isReconstructMsg(vbs::dispatcher::SubstrateReader* reader, const CacheChange_t* const change) {
    (void)reader;
    (void)change;
    return false;
}

void PDPListener::process_alive_data(ParticipantProxyData* old_data, ParticipantProxyData& new_data,
                                     GUID_t& writer_guid, RTPSReader* reader,
                                     std::unique_lock<std::recursive_mutex>& lock) {
    GUID_t participant_guid = new_data.m_guid;

    logDebug(SECURITY, "ParticipantProxyData guid:" << participant_guid << " writer:" << writer_guid
                                                    << " reader:" << reader->getGuid());
    if (old_data == nullptr) {
        // Create a new one when not found
        old_data = parent_pdp_->createParticipantProxyData(new_data, writer_guid);

        if (old_data != nullptr) {
            // Copy proxy to be passed forward before releasing PDP mutex
            ParticipantProxyData old_data_copy(*old_data);

            reader->getMutex().unlock();
            lock.unlock();
            // Assigning remote endpoints implies sending a DATA(p) to all matched and fixed readers, since
            // StatelessWriter::matched_reader_add marks the entire history as unsent if the added reader's
            // durability is bigger or equal to TRANSIENT_LOCAL_DURABILITY_QOS (TRANSIENT_LOCAL or TRANSIENT),
            // which is the case of ENTITYID_BUILTIN_SDP_PARTICIPANT_READER (TRANSIENT_LOCAL). If a remote
            // participant is discovered before creating the first DATA(p) change (which happens at the end of
            // BuiltinProtocols::initBuiltinProtocols), then StatelessWriter::matched_reader_add ends up marking
            // no changes as unsent (since the history is empty), which is OK because this can only happen if a
            // participant is discovered in the middle of BuiltinProtocols::initBuiltinProtocols, which will
            // create the first DATA(p) upon finishing, thus triggering the sent to all fixed and matched
            // readers anyways.
            parent_pdp_->assignRemoteEndpoints(&old_data_copy);
        } else {
            reader->getMutex().unlock();
            lock.unlock();
        }
    } else {
        old_data->updateData(new_data);
        old_data->isAlive = true;

        reader->getMutex().unlock();

        logDebug(RTPS_PDP_DISCOVERY, "Update participant " << old_data->m_guid << " at "
                                                           << "MTTLoc: " << old_data->metatraffic_locators
                                                           << " DefLoc:" << old_data->default_locators
                                                           << " old startTime: " << old_data->m_startTime.seconds()
                                                           << "s " << old_data->m_startTime.nanosec() << "ns "
                                                           << " new startTime: " << new_data.m_startTime.seconds()
                                                           << "s " << new_data.m_startTime.nanosec() << "ns ");

        if (parent_pdp_->updateInfoMatchesEDP()) {
            parent_pdp_->mp_EDP->assignRemoteEndpoints(*old_data, true);
        }

        // Copy proxy to be passed forward before releasing PDP mutex
        ParticipantProxyData old_data_copy(*old_data);

        lock.unlock();
        vbs::common::ParticipantListener* listener = parent_pdp_->getRTPSParticipant()->getListener();
        if (listener != nullptr) {
            {
                std::lock_guard<std::mutex> cb_lock(parent_pdp_->callback_mtx_);
                ParticipantDiscoveryInfo info(old_data_copy);
                info.status = ParticipantDiscoveryInfo::CHANGED_QOS_PARTICIPANT;
                listener->onParticipantDiscovery(parent_pdp_->getRTPSParticipant()->getUserRTPSParticipant(),
                                                 std::move(info));
            }
        }
    }

    // Take again the reader lock
    reader->getMutex().lock();
}

bool PDPListener::check_discovery_conditions(ParticipantProxyData& participant_data) {
    bool ret = true;
    uint32_t remote_participant_domain_id = participant_data.m_domain_id;

    // In PDPSimple, do not match if the participant is from a different domain.
    // If the domain id is unknown, it is assumed to be the same domain
    if (remote_participant_domain_id != parent_pdp_->getRTPSParticipant()->get_domain_id() &&
        remote_participant_domain_id != edds::dds::DOMAIN_ID_UNKNOWN) {
        logDebug(RTPS_PDP_DISCOVERY, "Received participant with different domain id ("
                                         << remote_participant_domain_id << ") than ours ("
                                         << parent_pdp_->getRTPSParticipant()->get_domain_id() << ")");
        ret = false;
    }

    return ret;
}

bool PDPListener::get_key(CacheChange_t* change) {
    return ParameterList::readInstanceHandleFromCDRMsg(change, edds::dds::PID_PARTICIPANT_GUID);
}

} /* namespace rtps */
} /* namespace ertps */
} /* namespace evbs */
